package com.bw.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.fun.HbaseSinkFunction;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.TimeZone;

public class Test2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<String> stream = env.addSource(MyKafkaUtil.getKafkaConsumer("tms_ods_yk5","test699"));

        //4)
        OutputTag<String> hbaseTag = new OutputTag<String>("hbase") {};
        SingleOutputStreamOperator<String> orderInfoDS = stream.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String s, ProcessFunction<String, String>.Context context, Collector<String> collector) throws Exception {
                JSONObject js = JSON.parseObject(s);
                // 维度表
                String[] tbs = {"user_info", "base_dic", "base_region_info", "express_courier"};
                List<String> tblist = Arrays.asList(tbs);
                if (tblist.contains(js.getString("table"))) {
                    // 侧流
                    context.output(hbaseTag, s);
                } else if ("order_info".equals(js.getString("table"))) {
                    System.out.println(js);
                    JSONObject data = js.getJSONObject("data");
                    String create_time = data.getString("create_time");
                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
                    sdf.setTimeZone(TimeZone.getTimeZone("GMT+8"));
                    long time = sdf.parse(create_time).getTime();
                    data.put("create_ts",time);
                    collector.collect(js.toJSONString());
                }
            }
        });
        DataStream<String> hbaseDS = orderInfoDS.getSideOutput(hbaseTag);
        hbaseDS.print();
//        hbaseDS.addSink(new HbaseSinkFunction());
        //5)
        orderInfoDS.print();
        //60020
        OutputTag<String> t60020 = new OutputTag<String>("t60020"){};
        //60030
        OutputTag<String> t60030 = new OutputTag<String>("t60030"){};
        //60050
        OutputTag<String> t60050 = new OutputTag<String>("t60050"){};
        //60060
        OutputTag<String> t60060 = new OutputTag<String>("t60060"){};
        //60070
        OutputTag<String> t60070 = new OutputTag<String>("t60070"){};
        //60080
        OutputTag<String> t60080 = new OutputTag<String>("t60080"){};
        //60999
        OutputTag<String> t60999 = new OutputTag<String>("t60999"){};
        SingleOutputStreamOperator<String> t60010DS = orderInfoDS.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String s, ProcessFunction<String, String>.Context context, Collector<String> collector) throws Exception {
                //{"database":"tms01","data":{},"type":"insert","table":"order_info"}
                JSONObject js = JSON.parseObject(s).getJSONObject("data");
                String stauts = js.getString("status");
                if ("60010".equals(stauts)) {
                    collector.collect(js.toString());
                } else if ("60020".equals(stauts)) {
                    context.output(t60020, js.toString());
                } else if ("60030".equals(stauts)) {
                    context.output(t60030, js.toString());
                } else if ("60050".equals(stauts)) {
                    context.output(t60050, js.toString());
                } else if ("60060".equals(stauts)) {
                    context.output(t60060, js.toString());
                } else if ("60070".equals(stauts)) {
                    context.output(t60070, js.toString());
                } else if ("60080".equals(stauts)) {
                    context.output(t60080, js.toString());
                } else if ("60999".equals(stauts)) {
                    context.output(t60999, js.toString());
                }
            }
        });

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop102:9092");
        properties.setProperty("group.id", "group1");
        FlinkKafkaProducer<String> p60010 = new FlinkKafkaProducer<>("dwd-order-detail-yk5-1", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60020 = new FlinkKafkaProducer<>("dwd-pay-detail-yk5", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60030 = new FlinkKafkaProducer<>("dwd-receive-detail-yk5", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60050 = new FlinkKafkaProducer<>("dwd-dispatch-detail-yk5", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60060 = new FlinkKafkaProducer<>("dwd-bound-detail-yk5", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60070 = new FlinkKafkaProducer<>("dwd-deliver-detail-yk5", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60080 = new FlinkKafkaProducer<>("dwd-sign-detail-yk5", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60999 = new FlinkKafkaProducer<>("dwd-cancel-detail-yk5", new SimpleStringSchema(), properties);
        t60010DS.addSink(p60010);
        t60010DS.getSideOutput(t60020).addSink(p60020);
        t60010DS.getSideOutput(t60030).addSink(p60030);
        t60010DS.getSideOutput(t60050).addSink(p60050);
        t60010DS.getSideOutput(t60060).addSink(p60060);
        t60010DS.getSideOutput(t60070).addSink(p60070);
        t60010DS.getSideOutput(t60080).addSink(p60080);
        t60010DS.getSideOutput(t60999).addSink(p60999);

        env.execute();
    }
}
